use std::{num::ParseFloatError, sync::Arc};

use anyhow::Result;
use mysql_async::Conn;
use reqwest::Client;
use serde::{Deserialize, Serialize};

use crate::{get_all_tables, get_data_size, get_tablet_by_name, AppConfig};

#[derive(Debug, Serialize, Deserialize)]
pub struct TableInfo {
    pub db_name: String,
    pub table_name: String,
    pub replication_counts: u32,
    pub data_size: f32,
    pub tablet_size_max: f32,
    pub tablet_size_min: f32,
    pub tablet_size_avg: f32,
    pub tablet_size_std: f32,
}

pub async fn get(dbs: &str, config: Arc<AppConfig>, pool: &mut Conn) -> Result<()> {
    let dbs = dbs.split(",");
    let mut tabs: Vec<TableInfo> = Vec::new();
    for ele in dbs.into_iter() {
        let db = ele.trim();
        let tables = get_all_tables(db, pool).await?;
        for ele in tables {
            fetch_all_table_infos(&ele.table_schema, &ele.table_name, &mut tabs, pool).await?;
        }
    }
    let json = serde_json::to_string(&tabs)?;
    push_to_starrocks(&json, config).await?;
    // println!("{}", json);
    Ok(())
}

async fn push_to_starrocks(json: &str, config: Arc<AppConfig>) -> Result<()> {
    let host = &config.db_config.host;
    let username = &config.db_config.username;
    let password = &config.db_config.password;
    let url = format!(
        "http://{}:8030/api/starrocks_audit_db__/starrocks_tablet_manager/_stream_load",
        host
    );
    let mut header = reqwest::header::HeaderMap::new();

    header.insert("format", "json".parse().unwrap());
    header.insert("strip_outer_array", "true".parse().unwrap());
    let client = Client::new();
    let response = client
        .post(&url)
        .basic_auth(username, Some(password))
        .headers(header)
        .body(json.to_string())
        .send()
        .await?;
    if response.status().is_success() {
        let response_text = response.text().await?;
        println!("push to starrocks success: {}", response_text);
    } else {
        let error_text = response.text().await?;
        eprintln!("Failed to import data: {}", error_text);
    }
    Ok(())
}

async fn fetch_all_table_infos(
    schema: &str,
    table: &str,
    tabs: &mut Vec<TableInfo>,
    pool: &mut Conn,
) -> Result<()> {
    // let mut tabs: Vec<TableInfo> = Vec::new();
    let partitions = get_data_size(schema, table, pool).await?;
    let data: String = partitions.get(0).unwrap().get(2).unwrap();
    let replica_count: u32 = partitions.get(0).unwrap().get(3).unwrap();
    let tablets = get_tablet_by_name(schema, table, pool).await?;
    let mut size: Vec<f32> = Vec::new();
    for ele in tablets {
        let data_size: String = ele.get(11).unwrap();
        if let Ok(ret) = analysis_data_size(data_size.clone()) {
            size.push(ret);
        } else {
            println!("{}:{}:{data_size:?}", schema, table);
        };
    }
    size.sort_by(|a, b| a.partial_cmp(b).unwrap());
    let min = size.first().unwrap_or(&0.0);
    let max = size.last().unwrap_or(&0.0);
    let sum: f32 = size.iter().sum();
    let avg = sum / size.len() as f32;
    let std = (size
        .clone()
        .iter()
        .map(|&x| (x - size.iter().sum::<f32>() / size.len() as f32).powi(2))
        .sum::<f32>()
        .sqrt()) as f32;
    let data = analysis_data_size(data).unwrap();
    let info = TableInfo::new(
        schema.to_string(),
        table.to_string(),
        replica_count,
        data,
        *max,
        *min,
        avg,
        std,
    );
    tabs.push(info);
    Ok(())
}

#[allow(unused)]
fn analysis_data_size(data_size: String) -> Result<f32, Error> {
    let data_size = data_size.trim(); // 去除前后空格

    let (number, unit) = if let Some(position) = data_size.find(char::is_alphabetic) {
        data_size.split_at(position)
    } else {
        (data_size, "B") // 默认单位为字节（B）
    };

    let number: f32 = number.trim().parse()?;

    let size_in_mb = match unit.trim() {
        "B" => number / 1024.0 / 1024.0, // 字节转换为MB
        "KB" => number / 1024.0,         // KB转换为MB
        "MB" => number,                  // MB保持不变
        "GB" => number * 1024.0,         // GB转换为MB
        _ => return Err(Error::InvalidUnit),
    };

    Ok((size_in_mb * 100.0).round() / 100.0)
}

#[allow(unused)]
#[derive(Debug)]
enum Error {
    ParseError(ParseFloatError),
    InvalidUnit,
}

impl From<ParseFloatError> for Error {
    fn from(err: ParseFloatError) -> Error {
        Error::ParseError(err)
    }
}

impl TableInfo {
    fn new(
        db_name: String,
        table_name: String,
        replication_counts: u32,
        data_size: f32,
        tablet_size_max: f32,
        tablet_size_min: f32,
        tablet_size_avg: f32,
        tablet_size_std: f32,
    ) -> Self {
        Self {
            db_name,
            table_name,
            replication_counts,
            data_size,
            tablet_size_max,
            tablet_size_min,
            tablet_size_avg,
            tablet_size_std,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_analysis_data_size() {
        let data_size = "0.000 ".to_string();
        let result = analysis_data_size(data_size);
        assert_eq!(result.unwrap(), 1.0 / 1024.0);
    }
}
